package sunyu.kit.kafka;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.slf4j.Logger;
import sunyu.kit.common.ExceptionKit;
import sunyu.kit.common.LogKit;

import java.util.Arrays;
import java.util.Properties;

/**
 * 消费者模版
 *
 * @author 孙宇
 */
public class KafkaConsumer {

    private static final Logger logger = LogKit.getLogger();

    private boolean runner = true;

    public KafkaConsumer(String servers,
                         String keyDeserializer,
                         String valueDeserializer,
                         String group,
                         String topic) {
        logger.info("初始化Kafka Consumer开始");
        logger.info("servers {}", servers);
        logger.info("keyDeserializer {}", keyDeserializer);
        logger.info("valueDeserializer {}", valueDeserializer);
        logger.info("group {}", group);
        logger.info("topic {}", topic);

        Properties props = new Properties();
        props.put("bootstrap.servers", servers);
        props.put("key.deserializer", keyDeserializer);
        props.put("value.deserializer", valueDeserializer);
        props.put("group.id", group);//不同ID 可以同时订阅消息
        /*
        props.put("enable.auto.commit", "false");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");*/
        try (Consumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(props);) {
            consumer.subscribe(Arrays.asList(topic.split(",")));//订阅多个TOPIC
            logger.info("初始化Kafka Consumer结束");
            while (runner) {
                ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
                for (ConsumerRecord<String, String> record : records) {
                    logger.debug("topic = {}, partition = {}, offset = {}, key = {}, value = {}",
                            record.topic(), record.partition(), record.offset(), record.key(), record.value());
                    /*consumer.commitSync();//同步*/
                }
            }
        } catch (Exception e) {
            logger.error(ExceptionKit.getStackTrace(e));
        }
    }

    public void setRunner(boolean runner) {
        this.runner = runner;
    }

}
